package com.ai.wxy.spring.storm.test.drpc;

import lombok.extern.slf4j.Slf4j;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.tuple.Fields;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Topology
 *
 * @author 石头
 * @Date 2019/12/18
 * @Version 1.0
 **/
@Slf4j
public class TiwtterReachTopology {
    // mock存储的URL
    public static Map<String, String> URLS_DB = new HashMap<>();
    // 模拟Who转发了URL
    public static Map<String, List<String>> TWEETERS_DB = new HashMap<String, List<String>>();
    // 模拟转发人员的粉丝人员
    public static Map<String, List<String>> FOLLOWERS_DB = new HashMap<String, List<String>>();
    /**
     * 初始化数据， 实际数据应该保存在Database / Mongon / Redis， 这里仅仅模拟学习
     */
    static {

        URLS_DB.put("1", "www.tiwtter.com");
        URLS_DB.put("2", "engineering.twitter.com/blog/5");
        URLS_DB.put("3", "tech.backtype.com/blog/123");

        TWEETERS_DB.put("1", Arrays.asList("sally", "bob", "tim", "george", "nathan"));
        TWEETERS_DB.put("2", Arrays.asList("adam", "david", "sally", "nathan"));
        TWEETERS_DB.put("3", Arrays.asList("tim", "mike", "john"));

        FOLLOWERS_DB.put("george", Arrays.asList("alice","adam", "jim", "chris", "jai"));
        FOLLOWERS_DB.put("sally",  Arrays.asList("bob", "tim", "alice", "adam", "jim", "chris", "jai"));
        FOLLOWERS_DB.put("bob",    Arrays.asList("tim", "nathan", "jim", "mary", "david", "vivian"));
        FOLLOWERS_DB.put("tim",    Arrays.asList("alex"));
        FOLLOWERS_DB.put("nathan", Arrays.asList("sally", "bob", "adam", "harry", "chris", "vivian", "emily", "jordan"));
        FOLLOWERS_DB.put("adam",   Arrays.asList("david", "carissa"));
        FOLLOWERS_DB.put("mike",   Arrays.asList("john", "bob"));
        FOLLOWERS_DB.put("john",   Arrays.asList("alice", "nathan", "jim", "mike", "bob"));
    }
    private Config config;
    private LinearDRPCTopologyBuilder linearDRPCTopologyBuilder;
    public TiwtterReachTopology(String functionName){
        config = new Config();
        linearDRPCTopologyBuilder = new LinearDRPCTopologyBuilder(functionName);
        linearDRPCTopologyBuilder.addBolt(new GetTweetersBolt(),2);
        linearDRPCTopologyBuilder.addBolt(new GetFollowersBolt(), 5).shuffleGrouping();
        //这一步是防止follower太多，内存不够，从而使用id,follower来进行拆分统计
        linearDRPCTopologyBuilder.addBolt(new PartialUniquerBatchBolt(), 6).fieldsGrouping(new Fields("Id", "Follower"));
        linearDRPCTopologyBuilder.addBolt(new CountAggregatorBatchBlot(), 3).fieldsGrouping(new Fields("Id"));
    }

    public LocalDRPC submitLocal(){
        try {
            LocalDRPC drpc = new LocalDRPC();
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("TiwtterReachTopology", config, linearDRPCTopologyBuilder.createLocalTopology(drpc));
            return drpc;
        } catch (Exception e) {
            log.error("------LocalDRPC.error--------",e);
        }
        return null;
    }
    public void submitRemote(){
        // 设置两个worker结点
        config.setNumWorkers(2);
        try {
            StormSubmitter.submitTopology("TiwtterReachTopology", config,linearDRPCTopologyBuilder.createRemoteTopology());
        }catch (Exception e) {
            log.error("------submitRemote.error--------",e);
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        TiwtterReachTopology tiwtterReachTopology = new TiwtterReachTopology("TiwtterReach");
        if (args == null || args.length == 0) {
            LocalDRPC localDRPC = tiwtterReachTopology.submitLocal();
            log.info("*****>>Please input a free flag to start reach function");

            for (String id : URLS_DB.keySet()) {
                log.info("*****>>The URL {} Reach Count :{}",URLS_DB.get(id),localDRPC.execute("TiwtterReach", URLS_DB.get(id)));
            }

            System.exit(1);
            return;
        }else {
            tiwtterReachTopology.submitRemote();
        }
    }
}
